package com.cloud.adrian.novel.search.task;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Time;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
import com.cloud.adrian.novel.book.dto.resp.BookEsRespDto;
import com.cloud.adrian.novel.common.constant.EsConstants;
import com.cloud.adrian.novel.search.manager.feign.BookFeignManager;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 小说数据从mysql同步到es
 * @Author: xiangguiliang
 * @version: 1.0
 * @since: 2024/05/26
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class BookToEsTask {
    private final ElasticsearchClient esClient;

    private final BookFeignManager bookFeignManager;

//    @SneakyThrows
//    @XxlJob("saveToEsJobHandler")
    public ReturnT<String> saveToEs() {
        try {
            long maxId = 0;
            for (;;) {
                List<BookEsRespDto> books = bookFeignManager.listEsBooks(maxId);
                if (books.isEmpty()) {
                    break;
                }
                BulkRequest.Builder br = new BulkRequest.Builder();

                for (BookEsRespDto book : books) {
                    br.operations(op -> op.index(
                            idx -> idx.index(EsConstants.BookIndex.INDEX_NAME)
                                      .id(book.getId().toString())
                                      .document(book))
                    ).timeout(Time.of(t -> t.time("10s")));
                    maxId = book.getId();
                }
                BulkResponse bulk = esClient.bulk(br.build());
                if (bulk.errors()) {
                    log.error("Bulk had error");
                    for (BulkResponseItem item : bulk.items()) {
                        if (item.error() != null) {
                            log.error(item.error().reason());
                        }
                    }
                }
            }
            return ReturnT.SUCCESS;
        } catch (Exception e) {
            return ReturnT.FAIL;
        }
    }
}
